/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

/***************************************************************************
 * Name: sock_session.cpp
 *
 * Purpose: socket session base API define and implementation
 *
 * Developer:
 *   wen.gu , 2023-11-01
 *
 * TODO:
 *
 ***************************************************************************/

#include "xbus/com/internal/sock_session.h"

#include "xbus/core/xtime.h"
#include "xbus/com/xmessage_impl.h"

#define GET_SEND_MSG_TIMEOUT_MS (10)
#define SOCK_RECV_TIMEOUT_MS (10)

namespace xbus::com::internal {
using namespace xbus::core;



static XResult<XMessageSp> SockRecv(XSocketId com_id, int32_t timeout_ms /*= core::WaitType::WaitForever*/) {
    XMessageHeader hdr = {0};
    //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
    auto ret = XSocketRecv(com_id, (uint8_t*)&hdr, sizeof(hdr), timeout_ms); //wait 10 ms,if nothing received

    //printf("[%s.%d]====>: ret: %d \n", __FUNCTION__, __LINE__, ret);
    if (!ret) {
        //todo process the error?   
        //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        return XResult<XMessageSp>::FromError(ret.error());
    }

    XSocketRecvResult res = ret.value();

    if (res.recv_size < static_cast<uint32_t>(sizeof(hdr))) {
        return XResult<XMessageSp>::FromError(XErrC::Undefined);
    }

    //printf("[%s.%d]====>:msg hdr: length: %d, start_code: 0x%02x, type: %d, protocol_version: %d\n", __FUNCTION__, __LINE__, 
    //hdr.length, hdr.start_code, hdr.type, hdr.protocol_version);
    if (XMessage::CheckMessageHeader(hdr)) {
        //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        XMessageSp msg_ptr = XMessageImpl::CreateMsg(hdr.length);
        //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        if (msg_ptr) {
            //printf("[%s.%d]====>:, length: %d, hdr_size:%d\n", __FUNCTION__, __LINE__, hdr.length, sizeof(hdr));
            ret = XSocketRecv(com_id, msg_ptr->payload(), msg_ptr->payload_size(), timeout_ms);//wait 40 ms,if nothing received
            //printf("[%s.%d]====>: ret: %d, payload_size(%d)\n", __FUNCTION__, __LINE__, ret, msg_ptr->payload_size());
            if (!ret) {
                //todo process the error?   
                //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
                return XResult<XMessageSp>::FromError(ret.error());
            }

            res = ret.value();

            if (res.recv_size < msg_ptr->payload_size()) {
                return XResult<XMessageSp>::FromError(XErrC::Undefined);
            }

            //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
            msg_ptr->set_msg_src(com_id);
            return XResult<XMessageSp>{msg_ptr};
        } else {
            //todo process this error?
            //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        }
    } else {
        //todo process this error?
        return XResult<XMessageSp>::FromError(XErrC::InvalidMessage);
    }   

    //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
    return XResult<XMessageSp>::FromError(XErrC::Undefined);
}


///////////////////////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////

SockSession::~SockSession() {
    /** todo something */
    stop();
}

SockSession::SockSession(XSocketId sock_id, const std::string& session_name)
    :XSession(sock_id, session_name) {
    /** todo something */
}

XResultCode SockSession::start(const XSessionStartParam& param) {
    if ((session_id_ == INVALID_XSOCKET_ID) || send_worker_ || recv_worker_) {
        return XResultCode::FromError(XErrC::InvalidStatus);
    }

    send_queue_ = XMessageQueueCreate(param.max_send_queue_size);
    recv_queue_ = XMessageQueueCreate(param.max_recv_queue_size);
    send_worker_ = std::make_unique<XThread>("OnSend:" + session_name_, [&](){ this->onSendWorker(); });
    recv_worker_ = std::make_unique<XThread>("OnRecv:" + session_name_, [&](){ this->onRecvWorker(); });

    if (!send_queue_ || !recv_queue_ || !send_worker_ || !recv_worker_) {
        send_queue_ = nullptr;
        recv_queue_ = nullptr;
        send_worker_.reset(nullptr);
        recv_worker_.reset(nullptr);
        return XResultCode::FromError(XErrC::InsufficientReousces);
    }

    is_running_ = true;
    if ((send_queue_->start() != XErrC::OK) || (recv_queue_->start() != XErrC::OK) || !send_worker_->start() || !recv_worker_->start()) {
        is_running_ = false;
        send_worker_->join();
        recv_worker_->join();
        send_queue_ = nullptr;
        recv_queue_ = nullptr;
        send_worker_.reset(nullptr);
        recv_worker_.reset(nullptr);
        return XResultCode::FromError(XErrC::Undefined);    
    }

    return { };
}

XResultCode SockSession::stop() {
    if (!send_worker_ || !recv_worker_) {
        return XResultCode::FromError(XErrC::InvalidStatus);
    }

    is_running_ = false;
    send_worker_->join();
    recv_worker_->join();
    send_queue_ = nullptr;
    recv_queue_ = nullptr;
    send_worker_.reset(nullptr);
    recv_worker_.reset(nullptr);
    session_id_ = INVALID_XSOCKET_ID;
    return { };
}

XResultCode SockSession::reset(XSocketId sock_id){ /**  first clean send and receive queue, then update sock id */
    if (!send_worker_ || !recv_worker_) {
        return XResultCode::FromError(XErrC::InvalidStatus);
    }

    send_queue_->stop();
    recv_queue_->stop();

    send_queue_->clear();
    recv_queue_->clear();

    session_id_ = sock_id;
    send_queue_->start();
    recv_queue_->start();

    return { };
}


XResultCode SockSession::send(XMessageSp msg) {
    if (!send_queue_) {
        return XResultCode::FromError(XErrC::InvalidStatus);
    }

    auto ret = send_queue_->push(msg);

    if (ret == XErrC::OK) {
        return { };
    }

    return XResultCode::FromError(ret);
}

XResult<XMessageSp> SockSession::recv(int32_t timeout_ms /*= core::WaitType::WaitForever*/) {
    if (!recv_queue_) {
        return XResult<XMessageSp>::FromError(XErrC::InvalidStatus);
    }    

    return recv_queue_->pop(timeout_ms);
}


void SockSession::onSendWorker() {
    while (is_running_) {
        auto msg_res = send_queue_->pop(GET_SEND_MSG_TIMEOUT_MS); 
        if (msg_res) {
            auto msg = msg_res.value();
            auto res = XSocketSend(session_id_, msg->data(), msg->length());

            if (!res) {
                //todo process error.
            }
        } else {
            auto errc = msg_res.error();
            switch (errc) {
            case XErrC::InvalidStatus: 
                core::SleepMs(GET_SEND_MSG_TIMEOUT_MS);
                break;
            case XErrC::Disconnected:
                //todo process error
                break;
            default:
                break;
            }
        }
    }
}

void SockSession::onRecvWorker() {
    while (is_running_) {
        auto res = SockRecv(session_id_, SOCK_RECV_TIMEOUT_MS);
        if (res) {
            auto msg = res.value();
            auto ret = recv_queue_->push(msg);

            if (ret == XErrC::OutOfRange) {
                //todo process error
            }

        } else {
            auto errc = res.error();
            if (errc == XErrC::Disconnected) {
                //todo process error
            }
            core::SleepMs(SOCK_RECV_TIMEOUT_MS);
        }
    }
}

} //namespace xbus::com::internal